package com.wx.task;

import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Vector;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;

/**
 * 基于信号量对象池
 * <p>
 * T 对象类型
 * R 函数返回值
 *
 * @author wang.xun
 * @date 2023/07/26
 */
@Slf4j
public class SemaphoreObjectPool<T, R> {

    private List<T> pool;

    final Semaphore semaphore;

    public SemaphoreObjectPool(int size, T t) {

        pool = new Vector<>(size);
        for (int i = 0; i < size; i++) {
            pool.add(t);
        }
        semaphore = new Semaphore(size);

    }

    R exec(Function<T, R> function) throws InterruptedException {
        T t = null;
        semaphore.acquire();
        try {
            // 从池中获取对象
            t = pool.get(0);
            // 删除对象
            pool.remove(t);
            // 将对象传入函数接口
            return function.apply(t);
        } finally {
            // 执行完毕 添加回对象池
            pool.add(t);
            // 释放信号量
            semaphore.release();
        }
    }

    public static void main(String[] args) {
        SemaphoreObjectPool objectPool = new SemaphoreObjectPool(5, new String("Pool"));


        ExecutorService executorService = Executors.newFixedThreadPool(10, new ThreadFactory() {
            AtomicInteger index = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r);
                thread.setName("exeObject - " + index.incrementAndGet());

                return thread;
            }
        });

        /**
         * 限流 最多五个对象会进行执行
         */
        for (int i = 0; i < 100; i++) {

            executorService.submit(() -> {
                try {
                    Object rs = objectPool.exec((t) -> {
                        log.info("t :{}", t);
                        try {
                            Thread.sleep(3000);
                        } catch (InterruptedException e) {
                            throw new RuntimeException(e);
                        }
                        return "1123";
                    });

                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            });
        }


    }

}
